package io.task.message.socket.task;

import io.task.message.socket.ChannelContext;
import io.task.message.socket.Config;
import io.task.message.socket.Packet;
import io.task.message.socket.lock.MapWithLock;
import io.task.message.socket.thread.pool.QueueModeSynchronousTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Setter
@Getter
public class HandleTask extends QueueModeSynchronousTask<Packet> {

    private ChannelContext channelContext = null;
    private Config config = null;

    private final AtomicLong synFailCount = new AtomicLong();

    public HandleTask(ChannelContext channelContext, Executor executor) {
        super(executor);
        this.channelContext = channelContext;
        config = channelContext.config;
        getMsgQueue();
    }

    /**
     * 处理packet
     */
    public void handler(Packet packet) {


        long start = System.currentTimeMillis();
        try {
            Integer synSeq = packet.getSynSeq();
            if (synSeq != null && synSeq > 0) {
                MapWithLock<Integer, Packet> syns = config.getWaitingResps();
                Packet initPacket = syns.remove(synSeq);
                if (initPacket != null) {
                    synchronized (initPacket) {
                        syns.put(synSeq, packet);
                        initPacket.notify();
                    }
                } else {
                    log.error("[{}]同步消息失败, synSeq is {}, 但是同步集合中没有对应key值", synFailCount.incrementAndGet(), synSeq);
                }
            } else {
                config.getTioHandler().handler(packet, channelContext);
            }
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        } finally {
            long end = System.currentTimeMillis();
            long iv = end - start;


        }
    }


    @Override
    public void runTask() {
        Packet packet = null;
        while ((packet = msgQueue.poll()) != null) {
            handler(packet);
        }
    }


    /**
     * The msg queue.
     */
    private final Queue<Packet> msgQueue = null;

    @Override
    public Queue<Packet> getMsgQueue() {

        return msgQueue;
    }

}
